/**
 *  WRMTransactionManager.java        22-ott-2009
 *   
 * Copyright (c) 2009 mentalsmash.org
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 * 
 *  http://www.apache.org/licenses/LICENSE-2.0
 * 
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 * 
 *  The use of the Apache License does not indicate that this project is
 *  affiliated with the Apache Software Foundation.
 */

package org.mentalsmash.tazio.transactions.management.wrm;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.mentalsmash.documentation.Concurrency;
import org.mentalsmash.tazio.cache.Cache;
import org.mentalsmash.tazio.commons.timer.MonitorTimePerformances;
import org.mentalsmash.tazio.net.ril.exceptions.UnavailableRemoteObjectException;
import org.mentalsmash.tazio.transactions.RollbackException;
import org.mentalsmash.tazio.transactions.Transaction;
import org.mentalsmash.tazio.transactions.TransactionStatus;
import org.mentalsmash.tazio.transactions.TransactionalSystemException;
import org.mentalsmash.tazio.transactions.UserTransactionManager;
import org.mentalsmash.tazio.transactions.log.TransactionsLog;
import org.mentalsmash.tazio.transactions.management.CommitmentAction;
import org.mentalsmash.tazio.transactions.management.NullTazioTransaction;
import org.mentalsmash.tazio.transactions.management.RootSetUpdate;
import org.mentalsmash.tazio.transactions.management.SystemTransactionManager;
import org.mentalsmash.tazio.transactions.management.TazioTransaction;
import org.mentalsmash.tazio.transactions.management.TransactionStateException;
import org.mentalsmash.tazio.transactions.management.TransactionalOperation;
import org.mentalsmash.tazio.transactions.management.ValidationAgent;
import org.mentalsmash.tazio.transactions.management.rm.RMTransactionsFacade;
import org.mentalsmash.tazio.transactions.modules.Committers;
import org.mentalsmash.tazio.transactions.modules.ReadOnly;
import org.mentalsmash.tazio.transactions.objects.RootSet;
import org.mentalsmash.tazio.transactions.objects.TransactionalObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.Singleton;

/**
 * The <tt>WRMTransactionManager</tt> models the transaction manager that owns
 * the {@link Transaction} objects in the context of a <i>Weak Replica Manager</i>
 * instance.
 * <p>
 * It provides both system-level operations that are not exposed to end-users
 * and the {@link UserTransactionManager} interface, that let clients interact
 * with the transactional system
 * </p>
 * <p>
 * It implements the <i>singleton</i> pattern, therefore there exists only an
 * instance of <tt>WRMTransactionManager</tt> per <i>Weak Replica Manager </i>
 * </p>
 * 
 * @author Andrea Reale
 * @version 0.0.1
 */
@Concurrency(concurrentAccess = true, specification = "This class is a singleton, therefore it may be accessed by mutiple "
	+ "concurrent threads. All the status of this class is final, therefore "
	+ "there are no problems of concurrent modifications. Nevertheless the "
	+ "CommitmentAction the committs the transaction must ensure to "
	+ "use the locks provided by the instance of "
	+ "WRMTransactionManager to make sure that only a thread at time"
	+ " are in the validation and in the commitment sections")
@Singleton
class WRMTransactionManager implements UserTransactionManager,
	SystemTransactionManager {

    private final static Logger log = LoggerFactory
	    .getLogger(WRMTransactionManager.class);

    // instance of TazioTransaction local to a thread
    private final ThreadLocal<TazioTransaction> currentTransaction = new ThreadLocal<TazioTransaction>();

    private final TransactionsLog transactionsLog;

    private final RMTransactionsFacade replicaManager;

    private final ValidationAgent localValidationAgent;

    private final RootSet rootSet;

    private int localRollbacks;

    // used to take snapshot of transactional objects local to threads
    // private final Snapshooter snapshooter;

    // used to execute commitment actions
    private final ExecutorService executor;

    // lock the validation fase
    private final Lock validationLock = new ReentrantLock();

    private final Lock changesLock = new ReentrantLock();

    private final Cache localCache;

    @Inject
    private Provider<TazioTransaction> rwTransactionsProvider;

    @Inject
    @ReadOnly
    private Provider<TazioTransaction> readOnlyTransactionsProvider;

    @Inject
    private Provider<RMValidator> validatorProvider;

    @Inject
    WRMTransactionManager(final RMTransactionsFacade rmProxy,
	    final ValidationAgent agent, final TransactionsLog txLog,
	    final RootSet rs, /* final Snapshooter snapshooter, */
	    final Cache localCache, @Committers
	    final ExecutorService executor) {

	this.replicaManager = rmProxy;
	this.localValidationAgent = agent;
	this.transactionsLog = txLog;
	this.rootSet = rs;
	this.localCache = localCache;

	// this.snapshooter = snapshooter;
	this.executor = executor;

	// try {
	// this.rootSet = this.replicaManager.getRootSet();
	// } catch (RemoteException e) {
	// log.error("RMI Exception",e);
	// throw new TransactionalSystemRuntimeException(e);
	// }

    }

    /* ************************************************************************ */
    /*
         * From org.mentalsmash.tazio.transactions.UserTransactionManager
         * interface
         */
    /* ************************************************************************ */

    /*
         * (non-Javadoc)
         * 
         * @see org.mentalsmash.tazio.transactions.UserTransactionManager#begin()
         */
    @Override
    public Transaction begin() {
	return this.begin(false);

    }

    /*
         * (non-Javadoc)
         * 
         * @see org.mentalsmash.tazio.transactions.UserTransactionManager#beginReadOnly()
         */
    @Override
    public Transaction beginReadOnly() {
	return this.begin(true);
    }

    private Transaction begin(final boolean readonly) {
	TazioTransaction transaction;
	// // TODO: Make the following line better
	// transaction = (readonly ? new ReadOnlyTransaction(this,
	// this.snapshooter, TID.generateTID())
	// : new ReadWriteTransaction(this, this.snapshooter, TID
	// .generateTID()));

	transaction = readonly ? readOnlyTransactionsProvider.get()
		: rwTransactionsProvider.get();

	long initialSeqno = -1;
	currentTransaction.set(transaction);

	transaction.stateTransition(TransactionStatus.BEGINNING);

	try {
	    initialSeqno = this.replicaManager.beginTransaction();
	} catch (final UnavailableRemoteObjectException e) {
	    log.error("Cannot retrieve remote initial seqno");
	    return null;
	} // remote

	transaction.setSeqNo(initialSeqno);
	transaction.stateTransition(TransactionStatus.RUNNING);

	this.transactionsLog.logTransaction(transaction);

	return transaction;
    }

    /*
         * (non-Javadoc)
         * 
         * @see org.mentalsmash.tazio.transactions.UserTransactionManager#getTransaction ()
         */
    @Override
    public Transaction getTransaction() {
	final TazioTransaction tx = currentTransaction.get();
	if (tx == null) {
	    return NullTazioTransaction.getInstance();
	} else {
	    return tx;
	}
    }

    /***********************************************************************
         * 
         * From org.mentalsmash.tazio.transactions.SystemTransactionManager
         * interface
         * 
         * *************************************************************************
         */

    /*
         * (non-Javadoc)
         * 
         * @see org.mentalsmash.tazio.transactions.SystemTransactionManager#commit()
         */
    @MonitorTimePerformances
    @Override
    public void commit() throws RollbackException, TransactionStateException,
	    TransactionalSystemException {
	// TODO: when RM is functioning change to WRMCommitmentAction()
	// RootSet rs;
	// try {
	// rs = this.replicaManager.getRootSet();
	// } catch (RemoteException e) {
	// log.error("RMI Exception",e);
	// throw new TransactionalSystemRuntimeException(e);
	// }
	final CommitmentAction action = new WRMCommitmentAction(
		(TazioTransaction) this.getTransaction(), this, this.rootSet,
		validatorProvider.get());
	final WRMCommitmentTask task = new WRMCommitmentTask(action, this);
	try {
	    this.executor.execute(task);
	    task.get();
	} catch (final InterruptedException e) {
	    log.error("Unexpexted exception while committing transaction {}",
		    getTransaction(), e);
	    throw new TransactionalSystemException(e);
	} catch (final ExecutionException e) {
	    final Throwable cause = e.getCause();
	    log.debug("Error while committing transaction {}: {}",
		    getTransaction(), cause);
	    if (cause instanceof RollbackException) {
		throw (RollbackException) cause;
	    } else if (cause instanceof TransactionStateException) {
		throw (TransactionStateException) cause;
	    } else {
		throw new TransactionalSystemException(cause);
	    }
	}
    }

    /*
         * (non-Javadoc)
         * 
         * @see org.mentalsmash.tazio.transactions.SystemTransactionManager#commitAsync()
         */
    @Override
    public Future<TransactionStatus> commitAsync()
	    throws TransactionStateException {
	// TODO: when RM is functioning change to WRMCommitmentAction()
	// RootSet rs = null;
	// try {
	// rs = this.replicaManager.getRootSet();
	// } catch (RemoteException e) {
	// log.error("RMI Exception",e);
	// throw new TransactionalSystemRuntimeException(e);
	// }
	final CommitmentAction action = new WRMCommitmentAction(
		(TazioTransaction) this.getTransaction(), this, this.rootSet,
		this.validatorProvider.get());
	final WRMCommitmentTask task = new WRMCommitmentTask(action, this);
	this.executor.execute(task);
	return task;
    }

    /*
         * (non-Javadoc)
         * 
         * @see org.mentalsmash.tazio.transactions.SystemTransactionManager#getStatus()
         */
    @Override
    public TransactionStatus getStatus() {
	final Transaction tx = getTransaction();
	return tx.getStatus();
    }

    /*
         * (non-Javadoc)
         * 
         * @seeorg.mentalsmash.tazio.transactions.SystemTransactionManager#
         * getTransactionsLog()
         */
    @Override
    public TransactionsLog getTransactionsLog() {
	return this.transactionsLog;
    }

    /*
         * (non-Javadoc)
         * 
         * @see org.mentalsmash.tazio.transactions.SystemTransactionManager#rollback()
         */
    @Override
    public void rollback() throws TransactionStateException {
	final TazioTransaction tx = (TazioTransaction) this.getTransaction();
	if (tx.getStatus() != TransactionStatus.RUNNING) {
	    log
		    .error("Transaction"
			    + tx
			    + " that wasn't in the RUNNING status was asked to rollbkack");
	    throw new TransactionStateException("Transaction " + tx
		    + " was not in the RUNNING status");
	}

	// cycling on write ops to intercept root set update operation and
	// update the cache properly

	for (TransactionalOperation op : tx.getOperations(RootSet.ROOT_SET_OID)) {
	    RootSetUpdate rsu = (RootSetUpdate) op;
	    //
	    if (!rsu.isRemoval()) {
		TransactionalObject wasToAdd = (TransactionalObject) rsu
			.getValue();
		localCache.removeObject(wasToAdd.getObjectIdentifier());
	    }
	}

	tx.stateTransition(TransactionStatus.ROLLED_BACK);
	clearThreadTransaction();
    }

    /*
         * (non-Javadoc)
         * 
         * @seeorg.mentalsmash.tazio.transactions.SystemTransactionManager#
         * getValidationAgent()
         */
    @Override
    public ValidationAgent getValidationAgent() {
	return this.localValidationAgent;
    }

    /*
         * Other methods
         */
    /**
         * Returns a reference to the local proxy of the <i>Replica Manager</i>
         * used by this <tt>WRMTransactionManager</tt>
         * 
         * @return a reference to the local proxy of the <i>Replica Manager</i>
         *         used by this <tt>WRMTransactionManager</tt>
         * @see RMTransactionsFacade
         */
    protected RMTransactionsFacade getReplicaManagerFacade() {
	return this.replicaManager;
    }

    protected Lock getValidationLock() {
	return this.validationLock;
    }

    protected Lock getChangesLock() {
	return this.changesLock;
    }

    public void clearThreadTransaction() {
	this.currentTransaction.set(null);
    }

    // used to shut down executor threads
    @Override
    protected void finalize() {
	log.debug("Finalizing committer threads");
	executor.shutdownNow();

    }

    //used to collect statistcs
    void incrementLocalRollbacks() {
	validationLock.lock();
	try {
	    this.localRollbacks++;
	} finally {
	    validationLock.unlock();
	}
    }
    
    public int getNumLocalRollbacks() {
	validationLock.lock();
	try {
	    return this.localRollbacks;
	} finally {
	    validationLock.unlock();
	}
    }

}
